package com.demo.realtime.job;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import com.demo.realtime.source.JbdcConnSingleton;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

/**
 * @Author zengqingdong
 * @ClassName PayMentJob
 * @Date 2020/8/10 23:49
 * @Version v1.0
 **/
public class PayMentJob {

    public static void main(String[] args) {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(streamEnv);
        streamEnv.enableCheckpointing(3000);
        CheckpointConfig checkpointConfig = streamEnv.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(6000);
        checkpointConfig.setMinPauseBetweenCheckpoints(500);
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        JdbcInputFormat.JdbcInputFormatBuilder jdbcInputFormatBuilder = JbdcConnSingleton.getInstance();
        JdbcInputFormat jdbcInputFormat = jdbcInputFormatBuilder.setQuery("select serviceCode,accountPeriod,productName,productCode,amt from sod_payment_bill_info where status='01'").finish();
        jdbcInputFormat.openInputFormat();
        DataStreamSource dataStreamSource = streamEnv.createInput(jdbcInputFormat);
        Table payMentBillTable = tEnv.fromDataStream(dataStreamSource).as("serviceCode","accountPeriod","productName","productCode","amt");
        Table payMentBillTable1 = tEnv.sqlQuery("select serviceCode,accountPeriod,productName,productCode,sum(amt) from" + payMentBillTable +" group by serviceCode,accountPeriod,productName,productCode");
        DataStream<Tuple2<Boolean, Row>>  dataStream = tEnv.toRetractStream(payMentBillTable1, Row.class);
        DataStream<String> stringDataStream = dataStream.flatMap(new FlatMapFunction<Tuple2<Boolean, Row>, String>() {
            @Override
            public void flatMap(Tuple2<Boolean, Row> elemnt, Collector<String> out) throws Exception {
                out.collect(elemnt.f1.toString());
            }
        });
        //插入es
    }
}
